package com.pipi.mq.rocketmq.base.producer;

import com.alibaba.fastjson.JSON;
import com.pipi.mq.rocketmq.base.MQException;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PreDestroy;
import java.nio.charset.Charset;

/**
 * All rights Reserved, Designed By www.maihaoche.com
 *
 * @Package com.golf.core.rocketmq
 * @author: 皮皮（pipi@maihaoche.com）
 * @date: 2019/5/15 10:18 AM
 * @Copyright: 2017-2020 www.maihaoche.com Inc. All rights reserved.
 * 注意：本内容仅限于卖好车内部传阅，禁止外泄以及用于其他的商业目
 */
public abstract class AbstractMQProducer {
    @Autowired
    private DefaultMQProducer producer;
    private String tag;
    private String topic;

    public String getTag() {
        return this.tag;
    }

    public void setTag(String tag) {
        this.tag = tag;
    }

    @PreDestroy
    public void destroyProducer() {
        if (this.producer != null) {
            Class var1 = AbstractMQProducer.class;
            synchronized(AbstractMQProducer.class) {
                if (this.producer != null) {
                    this.producer.shutdown();
                }
            }
        }

    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public String getTopic() {
        return this.topic;
    }

    private Message genMessage(String topic, String tag, String key, Object msgObj) {
        String str = JSON.toJSONString(msgObj);
        if (StringUtils.isEmpty(topic)) {
            if (StringUtils.isEmpty(this.getTopic())) {
                throw new RuntimeException("no topic defined to send this message");
            }

            topic = this.getTopic();
        }

        Message message = new Message(topic, str.getBytes(Charset.forName("utf-8")));
        if (!StringUtils.isEmpty(tag)) {
            message.setTags(tag);
        } else if (!StringUtils.isEmpty(this.getTag())) {
            message.setTags(this.getTag());
        }

//        if (StringUtils.isEmpty(key)) {
//            List<String> keys = (List) Arrays.stream(msgObj.getClass().getDeclaredFields()).filter((field) -> {
//                return field.isAnnotationPresent(MQKey.class);
//            }).map((field) -> {
//                try {
//                    String prefix = ((MQKey)field.getAnnotation(MQKey.class)).prefix();
//                    field.setAccessible(true);
//                    return prefix + field.get(msgObj);
//                } catch (Exception var3) {
//                    log.error("gen message key error : {}", var3.getLocalizedMessage());
//                    return "";
//                }
//            }).collect(Collectors.toList());
//            if (!CollectionUtils.isEmpty(keys)) {
//                message.setKeys(keys);
//            }
//        } else {
//            message.setKeys(key);
//        }
        message.setKeys(key);

        return message;
    }

    public void sendOneWay(String topic, String tag, String key, Object msgObj) {
        try {
            this.producer.sendOneway(this.genMessage(topic, tag, key, msgObj));
        } catch (Exception var6) {
            throw new MQException("消息发送失败，topic :" + topic + ",e:" + var6.getMessage());
        }
    }

    public void sendOneWay(Object msgObj) throws MQException {
        this.sendOneWay("", "", "", msgObj);
    }

    public void sendOneWay(String tag, Object msgObj) throws MQException {
        this.sendOneWay("", tag, "", msgObj);
    }

    public void sendOneWay(String tag, String key, Object msgObj) throws MQException {
        this.sendOneWay("", tag, key, msgObj);
    }

    public void synSend(String topic, String tag, String key, Object msgObj) throws MQException {
        try {
            if (null != msgObj) {
                SendResult sendResult = this.producer.send(this.genMessage(topic, tag, key, msgObj));
                this.doAfterSynSend(sendResult);
            }
        } catch (Exception var6) {
            throw new MQException("消息发送失败，topic :" + topic + ",e:" + var6.getMessage());
        }
    }

    public void synSend(Object msgObj) throws MQException {
        this.synSend("", "", "", msgObj);
    }

    public void synSend(String tag, Object msgObj) throws MQException {
        this.synSend("", tag, "", msgObj);
    }

    public void synSend(String tag, String key, Object msgObj) throws MQException {
        this.synSend("", tag, key, msgObj);
    }

    public void asynSend(String topic, String tag, String key, Object msgObj, SendCallback sendCallback) throws MQException {
        try {
            if (null != msgObj) {
                this.producer.send(this.genMessage(topic, tag, key, msgObj), sendCallback);
            }
        } catch (Exception var7) {
            throw new MQException("消息发送失败，topic :" + topic + ",e:" + var7.getMessage());
        }
    }

    public void asynSend(Object msgObj, SendCallback sendCallback) throws MQException {
        this.asynSend("", "", "", msgObj, sendCallback);
    }

    public void asynSend(String tag, Object msgObj, SendCallback sendCallback) throws MQException {
        this.asynSend("", tag, "", msgObj, sendCallback);
    }

    public void asynSend(String tag, String key, Object msgObj, SendCallback sendCallback) throws MQException {
        this.asynSend("", tag, key, msgObj, sendCallback);
    }

    /** @deprecated */
    public void sendMessage(Object msgObj) throws MQException {
        if (StringUtils.isEmpty(this.getTopic())) {
            throw new MQException("如果用这种方式发送消息，请在实例中重写 getTopic() 方法返回需要发送的topic");
        } else {
            this.sendOneWay("", "", "", msgObj);
        }
    }

    public void doAfterSynSend(SendResult sendResult) {
        System.out.println(JSON.toJSONString(sendResult));
    }
}
